package mostpopular.spark;

import mostpopular.User.UserLog;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.FilterFunction;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;

public class GetMostPopularSpark {

    public static void GetResult(JavaPairRDD<String, Integer> javaPairRDDrResult) {
        Map<String, Integer> ItemPairResult = javaPairRDDrResult.collectAsMap();
        List<Map.Entry<String, Integer>> entryList = new ArrayList<>(ItemPairResult.entrySet());
        entryList.sort(Collections.reverseOrder(Map.Entry.comparingByValue()));
        System.out.println(entryList.size());
        for (int i = 0; i < 100 && i < entryList.size(); i++) {
            System.out.println(entryList.get(i).getKey() + ": " + entryList.get(i).getValue());
        }
    }

    public static void main(String[] args) throws IOException {
        SparkSession spark = SparkSession.builder().appName("Get MostPopular").getOrCreate();
        String UserInfoFileName = "./src/main/resources/data_format1/data_format1/user_info_format1.csv";
        String UserLogFileName = "./src/main/resources/data_format1/data_format1/user_log_format1.csv";

        JavaPairRDD<String, String> UserInfoPairRDD = spark.read().textFile(UserInfoFileName).javaRDD()
                .mapToPair((PairFunction<String, String, String>) s -> {
                    String[] values = s.split(",", -1);
                    return new Tuple2<>(values[0], values[1]);
                });
        Map<String, String> UserInfoMap = UserInfoPairRDD.collectAsMap();

        Dataset<UserLog> UserLogDataset = spark.read().textFile(UserLogFileName)
                .filter((FilterFunction<String>) s -> s.charAt(0) <= '9' && s.charAt(0) >= '0')
                .map((MapFunction<String, UserLog>) UserLog::new, Encoders.bean(UserLog.class))
                .filter((FilterFunction<UserLog>) UserLog::isPopular);

        JavaPairRDD<String, Integer> ItemPairRDD = UserLogDataset.javaRDD()
                .mapToPair((PairFunction<UserLog, String, Integer>) userLog -> new Tuple2<>(userLog.getItemId(), 1))
                .reduceByKey(Integer::sum);

        GetResult(ItemPairRDD);

        JavaPairRDD<String, Integer> SellerParirRDD = UserLogDataset
                .filter((FilterFunction<UserLog>) userLog -> UserLog.isYoungPeople(UserInfoMap, userLog))
                .javaRDD()
                .mapToPair((PairFunction<UserLog, String, Integer>) userLog -> new Tuple2<>(userLog.getSellerId(), 1))
                .reduceByKey(Integer::sum);

        GetResult(SellerParirRDD);
    }

}

